package com.clei.netty.sync.server;

import com.clei.consts.NettyConstants;
import com.clei.netty.file.enums.FileSyncCmdTypeEnum;
import com.clei.netty.sync.dto.MsgBody;
import com.clei.utils.ByteUtil;
import com.clei.utils.FileUtil;
import com.clei.utils.PrintUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.io.File;
import java.io.RandomAccessFile;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * file sync server handler
 *
 * @author Y
 * @date 2024-08-14
 */
public class FileSyncServerHandler extends SimpleChannelInboundHandler<MsgBody> {

    /**
     * channel group
     */
    private static final ChannelGroup CHANNEL_GROUP = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**
     * client -> sqlCounter
     */
    private static final Map<Integer, AtomicInteger> CLIENT_SEQ_COUNTER = new ConcurrentHashMap<>();

    /**
     * 文件目录分隔符
     */
    private static final char FILE_SEPARATOR = '/';

    /**
     * 等待队列
     */
    private final static BlockingQueue<Integer> WAIT_QUEUE = new ArrayBlockingQueue<>(10);

    /**
     * 有新连接加入
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        PrintUtil.log("[{}]handlerAdded", ctx.channel().remoteAddress());
        Channel client = ctx.channel();
        // 添加到组内
        CHANNEL_GROUP.add(client);
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        PrintUtil.log("[{}]handlerRemoved", ctx.channel().remoteAddress());
        // Channel client = ctx.channel();
        // 关闭的channel会自动从channelGroup remove
        // 无需调用channelGroup.remove(client)
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        PrintUtil.log("[{}]channelActive", ctx.channel().remoteAddress());
        super.channelActive(ctx);
        // 初始化计数器
        CLIENT_SEQ_COUNTER.put(getClientId(ctx), new AtomicInteger());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        PrintUtil.log("[{}]channelInactive", ctx.channel().remoteAddress());
        super.channelInactive(ctx);
        // 移除计数器
        CLIENT_SEQ_COUNTER.remove(getClientId(ctx));
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        PrintUtil.log("[{}]channelRegistered", ctx.channel().remoteAddress());
        super.channelRegistered(ctx);
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        PrintUtil.log("[{}]channelUnregistered", ctx.channel().remoteAddress());
        super.channelUnregistered(ctx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        super.channelRead(ctx, msg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        PrintUtil.log("[{}]异常关闭", ctx.channel().remoteAddress(), cause);
        // 遇到异常就关闭连接
        ctx.close();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MsgBody msg) {
        byte cmd = msg.getCmd();
        FileSyncCmdTypeEnum cmdType = FileSyncCmdTypeEnum.getByCmdCode(cmd);
        try {
            PrintUtil.log("address : {}, msg : {}", ctx.channel().remoteAddress(), msg);
            switch (cmdType) {
                case REQ_SYNC:
                    Executors.newSingleThreadExecutor().execute(() -> handleSyncReq(ctx, msg));
                    break;
                case RES_ACK:
                    // 不再等待
                    WAIT_QUEUE.poll();
                    break;
                default:
            }
        } catch (Exception e) {
            PrintUtil.log("请求处理出错 address : {}, msg : {},", ctx.channel().remoteAddress(), msg, e);
        }
    }

    /**
     * handleSyncReq
     *
     * @param ctx ctx 频道上下文
     * @param msg 源消息
     */
    private void handleSyncReq(ChannelHandlerContext ctx, MsgBody msg) {
        byte[] bytes = msg.getData();
        String fileName = new String(bytes, NettyConstants.DEFAULT_CHARSET);
        File file = new File(fileName);
        // 文件不存在
        if (!file.exists()) {
            sendMsg(ctx, msg.getSeq(), FileSyncCmdTypeEnum.RES_FILE_NOT_EXISTS);
            return;
        }

        String parentDirectory = "";
        if (file.isFile()) {
            sendFileMsg(ctx, parentDirectory, file, true);
        } else {
            File[] files = file.listFiles();
            if (null != files) {
                for (File f : files) {
                    sendFileMsg(ctx, parentDirectory, f, f.isFile());
                }
            }
        }

        // 等待文件数据处理完成
        while (!WAIT_QUEUE.isEmpty()) {
            try {
                TimeUnit.SECONDS.sleep(1L);
            } catch (Exception e) {
                PrintUtil.log("sleep error ", e);
            }
        }
        // 结束消息
        sendMsg(ctx, msg.getSeq(), FileSyncCmdTypeEnum.REQ_FINISH);
        PrintUtil.log("sync end");
    }

    /**
     * 发送文件数据消息
     *
     * @param ctx             ctx 频道上下文
     * @param parentDirectory 父目录
     * @param file            File
     */
    private void sendFileMsg(ChannelHandlerContext ctx, String parentDirectory, File file, boolean isFile) {
        String fileName = file.getName();
        PrintUtil.log("当前处理文件 : {} {}", parentDirectory, fileName);
        byte[] fileNameBytes = fileName.getBytes(NettyConstants.DEFAULT_CHARSET);
        byte[] fileNameLengthBytes = ByteUtil.intTo2Byte(fileNameBytes.length);
        byte[] parentDirBytes = parentDirectory.getBytes(NettyConstants.DEFAULT_CHARSET);
        byte[] parentDirLengthBytes = ByteUtil.intTo2Byte(parentDirBytes.length);
        if (isFile) {
            //  4 = 2 + 2 = 文件名长度 + 父目录长度  8 = offset
            int baseLength = MsgBody.HEADER_LENGTH + 4 + fileNameBytes.length + parentDirBytes.length + 8;
            int limitFileLength = 1048576 - baseLength;
            long fileLength = file.length();
            if (fileLength > limitFileLength) {
                byte[] prefix = ByteUtil.merge(fileNameLengthBytes, fileNameBytes, parentDirLengthBytes, parentDirBytes);
                int offset = 0;
                byte[] fileDataBytes = new byte[limitFileLength];
                try (RandomAccessFile raf = new RandomAccessFile(file, "r")) {
                    // 分块处理
                    while (offset < fileLength) {
                        long diff = fileLength - offset;
                        int readLength = limitFileLength;
                        if (diff < limitFileLength) {
                            readLength = (int) diff;
                            fileDataBytes = new byte[readLength];
                        }
                        // 跳到指定位置
                        raf.seek(offset);
                        raf.read(fileDataBytes);
                        // 文件名 + 父目录 + 文件数据
                        byte[] data = ByteUtil.merge(prefix, ByteUtil.longTo8Byte(offset), fileDataBytes);
                        MsgBody msgBody = new MsgBody(getSeq(ctx), FileSyncCmdTypeEnum.REQ_FILE_DATA.getCmdCode(), data);
                        sendFileMsg(ctx, msgBody);
                        offset = offset + readLength;
                    }
                } catch (Exception e) {
                    PrintUtil.log("读取文件出错 file : {} {}", parentDirectory, fileName, e);
                }
            } else {
                // 文件名 + 父目录 + 文件数据
                byte[] fileDataBytes = FileUtil.toBytes(file);
                byte[] data = ByteUtil.merge(fileNameLengthBytes, fileNameBytes, parentDirLengthBytes, parentDirBytes, ByteUtil.longTo8Byte(0), fileDataBytes);
                MsgBody msgBody = new MsgBody(getSeq(ctx), FileSyncCmdTypeEnum.REQ_FILE_DATA.getCmdCode(), data);
                sendFileMsg(ctx, msgBody);
            }
        } else {
            // 文件名 + 父目录 + offset
            byte[] data = ByteUtil.merge(fileNameLengthBytes, fileNameBytes, parentDirLengthBytes, parentDirBytes);
            MsgBody msgBody = new MsgBody(getSeq(ctx), FileSyncCmdTypeEnum.REQ_FILE_DIRECTORY.getCmdCode(), data);
            sendFileMsg(ctx, msgBody);
        }

        // 目录下文件处理
        if (!isFile) {
            File[] files = file.listFiles();
            if (null != files) {
                parentDirectory = parentDirectory + FILE_SEPARATOR + fileName;
                for (File f : files) {
                    sendFileMsg(ctx, parentDirectory, f, f.isFile());
                }
            }
        }
    }

    /**
     * 发送文件数据消息
     *
     * @param ctx     ctx 频道上下文
     * @param msgBody MsgBody
     */
    private void sendFileMsg(ChannelHandlerContext ctx, MsgBody msgBody) {
        // 同步等待 只能同时处理队列长度的任务数 可以慢一点 不能超负载
        try {
            WAIT_QUEUE.put(msgBody.getSeq());
        } catch (Exception e) {
            PrintUtil.log("wait error ", e);
        }
        sendMsg(ctx, msgBody);
    }

    /**
     * 获取用户id
     *
     * @param ctx 频道上下文
     * @return 用户id
     */
    private Integer getClientId(ChannelHandlerContext ctx) {
        return getClientId(ctx.channel());
    }

    /**
     * 获取用户id
     *
     * @param channel 频道
     * @return 用户id
     */
    private Integer getClientId(Channel channel) {
        return channel.hashCode() + channel.remoteAddress().toString().hashCode();
    }

    /**
     * 获取序号
     *
     * @param ctx ctx 频道上下文
     * @return seq
     */
    private int getSeq(ChannelHandlerContext ctx) {
        return getSeq(ctx.channel());
    }

    /**
     * 获取序号
     *
     * @param channel 频道
     * @return seq
     */
    private int getSeq(Channel channel) {
        Integer clientId = getClientId(channel);
        AtomicInteger seqCounter = CLIENT_SEQ_COUNTER.get(clientId);
        return seqCounter.incrementAndGet();
    }

    /**
     * 发送消息
     *
     * @param ctx ctx 频道上下文
     * @param ack 回应 对应序号 从1开始
     * @param cmd 命令类型
     */
    private void sendMsg(ChannelHandlerContext ctx, int ack, FileSyncCmdTypeEnum cmd) {
        sendMsg(ctx, new MsgBody(getSeq(ctx), ack, cmd.getCmdCode(), NettyConstants.EMPTY));
    }

    /**
     * 发送消息
     *
     * @param msgBody MsgBody
     */
    private void sendMsg(ChannelHandlerContext ctx, MsgBody msgBody) {
        ctx.writeAndFlush(msgBody);
    }
}
